#include "scheduler.h"

static bool debug = false;

namespace lzp {

// 当前线程的调度器，同一个调度器下的所有线程指向同一个调度器实例
static thread_local Scheduler* t_scheduler = nullptr;
// 当前线程的调度协程，每个线程都独有一份，包括caller线程
//static thread_local Fiber* t_scheduler_fiber = nullptr;

/**
 * @brief 获取当前线程调度器指针
 */
Scheduler* Scheduler::GetThis() {
    return t_scheduler;
}

/**
 * @brief 获取当前线程的主协程
 */
void Scheduler::SetThis() {
    t_scheduler = this;
}

/**
 * @brief 创建调度器
 * @param[in] threads 线程数
 * @param[in] use_caller 是否将当前线程也作为调度线程
 * @param[in] name 名称
 */
Scheduler::Scheduler(size_t threads, bool use_caller, const std::string& name) : m_useCaller(use_caller), m_name(name) {
    assert(threads > 0 && Scheduler::GetThis() == nullptr);
    
    SetThis();

    Thread::SetName(m_name);

    // 使用当前线程为调度线程，需要额外创建调度协程
    if (use_caller) {
        --threads;
        // 创建主协程
        Fiber::GetThis();

         /**
          * 在user_caller为true的情况下，初始化caller线程的调度协程
          * caller线程的调度协程不会被调度器调度，⽽且，caller线程的调度协程停⽌时，应该返回caller线
          * 程的主协程
          */
        // 创建调度协程，false -> 该调度协程退出后将返回主协程
        m_schedulerFiber.reset(new Fiber(std::bind(&Scheduler::run, this), 0, false));

        Thread::SetName(m_name);
        // t_scheduler_fiber = m_schedulerFiber.get();
        m_rootThread = Thread::GetThreadId();
        m_threadIds.push_back(m_rootThread);
    }
    m_threadCount = threads;
	if(debug) std::cout << "Scheduler::Scheduler() success\n";
}

/**
 * @brief 析构函数
 */
Scheduler::~Scheduler() {
    assert(stopping() == true);
    if (GetThis() == this) {
        t_scheduler = nullptr;
    }
    if(debug) std::cout << "Scheduler::~Scheduler() success\n";
}

/**
 * @brief 启动调度器
 */
void Scheduler::start() {
    std::lock_guard<std::mutex> lock(m_mutex);
    if (m_stopping) {
        std::cerr << "Scheduler is stopped" << std::endl;
		return;
    }

    // 启动时线程池为空，使用resize先分调整容器容量
    assert(m_threads.empty());
    m_threads.resize(m_threadCount);
    // 依次创建线程，每个线程对应一个调度协程
    for (size_t i = 0; i < m_threadCount; ++i) {
        m_threads[i].reset(new Thread(std::bind(&Scheduler::run, this), m_name + "_" + std::to_string(i)));
        m_threadIds.push_back(m_threads[i]->getId());
    }
    if(debug) std::cout << "Scheduler::start() success\n";
}

/**
 * @brief 协程调度函数
 */
void Scheduler::run() {
    int thread_id = Thread::GetThreadId();
    if(debug) std::cout << "Schedule::run() starts in thread: " << thread_id << std::endl;

    SetThis();

    // 运行在新创建的线程 -> 需要创建主协程，thread_local
    if (thread_id != m_rootThread) {
        Fiber::GetThis();
    }

    std::shared_ptr<Fiber> idle_fiber = std::make_shared<Fiber>(std::bind(&Scheduler::idle, this));
    ScheduleTask task;

    while (true) {
        task.reset();
        bool tickle_me = false; // 是否tickle其他线程进⾏任务调度

        // 遍历任务队列，找到合适的任务执行
        {
            std::lock_guard<std::mutex> lock(m_mutex);
            auto it = m_tasks.begin();
            
            while (it != m_tasks.end()) {
                // 当任务指定了线程运行，并且不是当前线程的时候，就跳过，并且标记需要通知其他线程进行调度
                if (it->thread != -1 && it->thread != thread_id) {
                    ++it;
                    tickle_me = true;
                    continue;
                }

                // 取出任务
                assert(it->fiber || it->cb);
                task = *it;
                m_tasks.erase(it);
                ++m_activeThreadCount;
                break;
            }
            tickle_me = tickle_me || (it != m_tasks.end());
        }

        if (tickle_me) {
            tickle();
        }

        // 执行任务，切换到协程任务函数执行
        if (task.fiber) {
            {
                std::lock_guard<std::mutex> lock(task.fiber->m_mutex);
                if (task.fiber->getState() != Fiber::TERM) {
                    // 切换到协程任务函数执行
                    task.fiber->resume();
                }
            }
            --m_activeThreadCount;
            task.reset();
        }
        // 创建协程并切换到协程任务函数执行
        else if (task.cb) {
            std::shared_ptr<Fiber> cb_fiber = std::make_shared<Fiber>(task.cb);
            {
                std::lock_guard<std::mutex> lock(cb_fiber->m_mutex);
                cb_fiber->resume();
            }
            --m_activeThreadCount;
            task.reset();
        }
        // 无任务，则执行空闲函数
        else {
            // 系统关闭 -> idle协程将从死循环跳出并结束 -> 此时的idle协程状态为TERM -> 再次进入将跳出循环并退出run()
            if (idle_fiber->getState() == Fiber::TERM) {
                if(debug) std::cout << "Schedule::run() ends in thread: " << thread_id << std::endl;
                break;
            }
            ++m_idleThreadCount;
            idle_fiber->resume();
            --m_idleThreadCount;
        }
    }
}

/**
 * @brief 停⽌调度器，等所有调度任务都执⾏完了再返回
 */
void Scheduler::stop() {
    if(debug) std::cout << "Schedule::stop() starts in thread: " << Thread::GetThreadId() << std::endl;
    if (stopping()) {
        return;
    }
    m_stopping = true;

    // 如果use caller，那只能由caller线程发起stop
    if (m_useCaller) {
        assert(GetThis() == this);
    }

    for (size_t i = 0; i < m_threadCount; i++) {
		tickle();
	}

	if (m_schedulerFiber) {
		tickle();
	}

    // 在use caller情况下，调度器协程结束时，应该返回caller协程
	if(m_schedulerFiber) {
		m_schedulerFiber->resume();
		if(debug) std::cout << "m_schedulerFiber ends in thread:" << Thread::GetThreadId() << std::endl;
	}

	std::vector<std::shared_ptr<Thread>> thrs;
	{
		std::lock_guard<std::mutex> lock(m_mutex);
		thrs.swap(m_threads);
	}

	for(auto &i : thrs) {
		i->join();
	}
	if(debug) std::cout << "Schedule::stop() ends in thread:" << Thread::GetThreadId() << std::endl;
}

/**
 * @brief 通知协程调度器有任务了
 */
void Scheduler::tickle() {

}

/**
 * @brief ⽆任务调度时执⾏idle协程
 */
void Scheduler::idle() {
	while(!stopping()) {
		if(debug) std::cout << "Scheduler::idle(), sleeping in thread: " << Thread::GetThreadId() << std::endl;	
		sleep(1);	
		Fiber::GetThis()->yield();
	}
}

/**
 * @brief 返回是否可以停⽌
 */
bool Scheduler::stopping() {
    std::lock_guard<std::mutex> lock(m_mutex);
    return m_stopping && m_tasks.empty() && m_activeThreadCount == 0;
}

}
